package cn.xuexiyuan.flinkstudy.source;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-3-23
 * @Version 1.0
 **/
public class SourceDemo03_Socket {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<String> ds1 = env.socketTextStream("localhost", 6666);

        // 2.transformation
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = ds1.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
            for (String token : s.split("\\W+")) {
                collector.collect(Tuple2.of(token, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(d -> d.f0).sum(1);

        // 3.sink
        result.print();

        // 4.excute
        env.execute("SourceDemo03_Socket");
    }
}
